package cn.lsj.lock.core;

import cn.lsj.lock.client.ZKClient;
import cn.lsj.lock.client.ZKClientFactory;
import cn.lsj.lock.config.DataSource;
import cn.lsj.lock.config.LockDefinition;
import org.apache.curator.RetryPolicy;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 可重入的公平非阻塞锁
 */
public class ReentrantLock implements Lock {

    private static final ExponentialBackoffRetry DEFAULT_RETRY_POLICY = new ExponentialBackoffRetry(1000, 3);

    private final RetryPolicy retryPolicy;

    private final DataSource dataSource;

    private final LockDefinition properties;

    // 重入计数
    private final AtomicInteger lockCount = new AtomicInteger(0);

    // 本地线程
    private Thread thread;

    private ZKClient client;

//    /**
//     * 临时节点路径
//     */
//    private String lockedPath;

//    /**
//     * 临时有序节点的编号节点
//     */
//    private String lockedShortPath;

//    /**
//     * 被监听的前一个节点
//     */
//    private String priorPath;

    private final ConcurrentMap<Thread, String> lockedPathMap = new ConcurrentHashMap<Thread, String>();

    private final ConcurrentMap<Thread, String> priorPathMap = new ConcurrentHashMap<Thread, String>();

//    private final ConcurrentMap<Thread, AtomicInteger> countMap = new ConcurrentHashMap<Thread, AtomicInteger>();

    public ReentrantLock(DataSource dataSource, LockDefinition properties) {
        this(dataSource, properties, DEFAULT_RETRY_POLICY);
    }

    public ReentrantLock(DataSource dataSource, LockDefinition properties, RetryPolicy retryPolicy) {
        this.dataSource = dataSource;
        this.properties = properties;
        this.retryPolicy = retryPolicy;
        init();
    }

    private void init() {
        this.client = ZKClientFactory.getInstance(dataSource, retryPolicy);
        if (!this.client.isNodeExist(this.properties.getZkPath())) {
            this.client.createNode(this.properties.getZkPath(), null);
        }
    }

    public boolean lock() {
        if(Thread.currentThread().equals(thread)) {
            lockCount.incrementAndGet();
            return true;
        }
        while (true) {
            if (tryLock()) {
                thread = Thread.currentThread();
                lockCount.incrementAndGet();
                return true;
            }
            try {
                await();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    public synchronized boolean unlock() {
        if (!thread.equals(Thread.currentThread())) {
            return false;
        }
        int newLockCount = lockCount.decrementAndGet();
        if (newLockCount < 0) {
            throw new IllegalMonitorStateException("重入锁计数不可为负数" );
        }
        // 是否剩余重入次数
        if (newLockCount != 0) {
            return true;
        }
        // 到这一步，意味着lockCount已经为0，可以删除临时节点了
        try{
            if(client.isNodeExist(properties.getZkPath())) {
                client.deleteNode(lockedPathMap.get(thread));
            }
        } catch (Exception e) {
            return false;
        } finally {
            lockedPathMap.remove(thread);
            priorPathMap.remove(thread);
        }
        return true;
    }

    protected boolean tryLock() {
        String lockedPath = lockedPathMap.get(Thread.currentThread());
        if (null == lockedPath || !client.isNodeExist(lockedPath)) {
            lockedPathMap.put(Thread.currentThread(), lockedPath = client.createEphemeralSeqNode(getLockPrefix()));
        }

        // 取得加锁的排队编号
        String lockedShortPath = getShorPath(lockedPath);
        List<String> waiters = getWaiters();
        // 如果自己是所有等待锁中的第一个，则获得锁
        if (checkLocked(waiters, lockedShortPath)) {
            return true;
        }

        // 当前线程节点是否在排队
        int index = Collections.binarySearch(waiters, lockedShortPath);
        if(index < 0) {
            throw new NullPointerException("可能网络抖动，连接断开，临时节点失效");
        }

        // waiters最后面的节点写入map，用来监听
        priorPathMap.put(Thread.currentThread(), getLockPrefix() + waiters.get(index - 1));

        return false;
    }

    private String getShorPath(String lockedPath) {
        int index = lockedPath.lastIndexOf(getLockPrefix());
        if (index >= 0) {
            index += properties.getZkPath().length() + 1;
            return index <= lockedPath.length() ? lockedPath.substring(index) : "";
        }
        throw new NullPointerException("获取节点编号失败，lockedPath=" + lockedPath);
    }

    private String getLockPrefix() {
        return properties.getZkPath() + "/";
    }

    /**
     * 判断首节点是否是当前线程的节点
     * @param waiters
     * @return
     */
    private boolean checkLocked(List<String> waiters, String lockedShortPath) {
        // 升序
        Collections.sort(waiters);
        // 如果是第一个，代表自己已经获得了锁
        return lockedShortPath.equals(waiters.get(0));
    }


    //线程排队实现
    private boolean await() throws Exception {
        String priorPath = priorPathMap.get(Thread.currentThread());
        if (null == priorPath) {
            throw new NullPointerException("prior_path error");
        }

        final CountDownLatch latch = new CountDownLatch(1);

        // 删除事件
        Watcher w = watchedEvent -> {
            // 监测到前一个节点发生变化，接下来就可以唤起等待线程，重新尝试获取锁
            latch.countDown();
        };

        try{
            // 监听前一个节点的删除时间
            client.watcher(w, priorPath);
        } catch (KeeperException.NoNodeException e) {
            e.printStackTrace();
            return false;
        }

        return latch.await(properties.getTimeout(), TimeUnit.MILLISECONDS);
    }


    /**
     * 从zookeeper中拿到所有等待节点
     */
    protected List<String> getWaiters() {
        return client.getWaiters(properties.getZkPath());
    }

}
